#ifndef __ZHONG_RABBITMQ__
#define __ZHONG_RABBITMQ__
#include <amqpcpp.h>
#include <ev.h>
#include <amqpcpp/libev.h>
#include <openssl/ssl.h>
#include <openssl/opensslv.h>
#include <iostream>
#include <functional>
#include "logger.hpp"

// namespace zhong
// {
//     // 基于AMQP封装MessageQueue通信的客户端
//     namespace rabbitmq
//     {
       
//     }
// }

namespace bite_im
{
     class MQClient
        {
#define default_routing_key "routing_key"
#define default_consumer_tag "consumer"
        public:
            using ptr = std::shared_ptr<MQClient>;
            // 外部传入给consume函数，作为用户实现的consume的消费回调函数
            using MessageCallback = std::function<void(const char *, size_t)>;
            MQClient(const std::string &user,
                     const std::string &password,
                     const std::string &host)
            {
                _loop = EV_DEFAULT;
                _hander = std::make_unique<AMQP::LibEvHandler>(_loop);
                // "amqp://root:123456@127.0.0.1:5672/"
                std::string addr_url = "amqp://" + user + ":" + password + "@" + host + "/";
                AMQP::Address address(addr_url.c_str()); // 是否要c_str
                _conn = std::make_unique<AMQP::TcpConnection>(_hander.get(), address);
                _channel = std::make_unique<AMQP::TcpChannel>(_conn.get());

                // 创建一个执行流来执行ev_run
                _loop_thread = std::thread([this]()
                                           { ev_run(_loop, 0); });
            }
            ~MQClient()
            {
                // 停止ev_run
                // 当我们析构时需要通知ev进行关闭执行流
                struct ev_async async_watcher;
                ev_async_init(&async_watcher, watcherCallback);
                ev_async_start(_loop, &async_watcher);
                ev_async_send(_loop, &async_watcher);
                _loop_thread.join();
                _loop = nullptr;
            }
            // 声明交换机和队列并进行绑定
            void declareComponents(const std::string &exchange_name,
                                   const std::string &queue_name,
                                   const std::string &routing_key = default_routing_key,
                                   AMQP::ExchangeType exchange_type = AMQP::ExchangeType::direct)
            {
                // 声明交换机
                _channel->declareExchange(exchange_name, exchange_type)
                    .onError([exchange_name](const char *message)
                             {
                                        LOG_ERROR("声明交换机[{}]失败，原因[{}]", exchange_name, message);
                                        exit(0); })
                    .onSuccess([exchange_name]()
                               { LOG_DEBUG("声明交换机[{}]成功", exchange_name); });
                // 声明队列
                _channel->declareQueue(queue_name)
                    .onError([queue_name](const char *message)
                             {
                                        LOG_ERROR("声明队列[{}]失败，原因[{}]", queue_name, message);
                                        exit(0); })
                    .onSuccess([queue_name]()
                               { LOG_DEBUG("声明队列[{}]成功", queue_name); });
                // 针对交换机和队列进行绑定
                _channel->bindQueue(exchange_name, queue_name, routing_key)
                    .onError([exchange_name, queue_name](const char *message)
                             {
                                        LOG_ERROR("绑定交换机[{}]-队列[{}]失败，原因[{}]", exchange_name, queue_name, message);
                                        exit(0); })
                    .onSuccess([exchange_name, queue_name]()
                               { LOG_DEBUG("绑定交换机[{}]-队列[{}]成功", exchange_name, queue_name); });
            }
            bool publish(const std::string &exchange_name,
                         const std::string &message,
                         const std::string &routing_key = default_routing_key)
            {
                bool ret = _channel->publish(exchange_name, routing_key, message);
                if (ret == false)
                {
                    LOG_INFO("向交换机[{}]发布消息失败", exchange_name);
                    return false;
                }
                return true;
            }
            void consume(const std::string &queue_name,
                         const std::string &consume_tag,
                         const MessageCallback &cb)
            {
                _channel->consume(queue_name, consume_tag) // 返回值 DeferredConsumer
                    .onReceived([this, cb](const AMQP::Message &message, uint64_t deliveryTag, bool redelivered)
                                { 
                                            cb(message.body(), message.bodySize()); 
                                            _channel->ack(deliveryTag); })
                    .onError([queue_name](const char *message)
                             {
                                        LOG_ERROR("订阅队列[{}]失败，原因[{}]", queue_name, message);
                                        exit(0); }); // 返回值是 AMQP::Deferred
            }

        private:
            static void watcherCallback(struct ev_loop *loop, ev_async *watcher, int32_t revents)
            {
                ev_break(loop, EVBREAK_ALL);
            }

        private:
            struct ev_loop *_loop;
            std::unique_ptr<AMQP::LibEvHandler> _hander;
            std::unique_ptr<AMQP::TcpConnection> _conn;
            std::unique_ptr<AMQP::TcpChannel> _channel;
            std::thread _loop_thread;
        };
}

#endif